package cn.saltybe.analysis.analyze;

import cn.saltybe.analysis.service.PageViewServiceImpl;
import cn.saltybe.analysis.util.RemoteExecute;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.stereotype.Component;

import java.text.SimpleDateFormat;
import java.util.Date;

@Component
public class PageViewAnalyze {
    private static PageViewServiceImpl pageViewService;

    @Autowired
    public void setNewHouseService(PageViewServiceImpl pageViewService) {
        this.pageViewService = pageViewService;
    }

    private static String datasourceUrl;
    private static String datasourceUsername;
    private static String datasourcePassword;
    private static String mapreduceJar;
    private static String mapreduceClass;
    private static String mapreduceInput;
    private static String mapreduceOutput;

    @Value("${spring.datasource.url}")
    public void setDatasourceUrl(String datasourceUrl) {
        this.datasourceUrl = datasourceUrl;
    }

    @Value("${spring.datasource.username}")
    public void setDatasourceUsername(String datasourceUsername) {
        this.datasourceUsername = datasourceUsername;
    }

    @Value("${spring.datasource.password}")
    public void setDatasourcePassword(String datasourcePassword) {
        this.datasourcePassword = datasourcePassword;
    }

    @Value("${mapreduce.user-view.jar}")
    public void setMapreduceJar(String mapreduceJar) {
        this.mapreduceJar = mapreduceJar;
    }

    @Value("${mapreduce.page-view.class}")
    public void setMapreduceClass(String mapreduceClass) {
        this.mapreduceClass = mapreduceClass;
    }

    @Value("${mapreduce.page-view.input}")
    public void setMapreduceInput(String mapreduceInput) {
        this.mapreduceInput = mapreduceInput;
    }

    @Value("${mapreduce.page-view.output}")
    public void setMapreduceOutput(String mapreduceOutput) {
        this.mapreduceOutput = mapreduceOutput;
    }

    private static String toDateString(Date date) {
        return new SimpleDateFormat("yy-MM-dd").format(date);
    }

    public static void analyze() {
        analyze(new Date());
    }

    public static void analyze(Date date) {
        System.out.println("* 执行 MapReduce 任务，对 页面访问量 进行分析");
        RemoteExecute re = RemoteExecute.getInstance();
        String out;
        String command;

        command = "hadoop dfs -rm -r \"" + mapreduceOutput + "/" + toDateString(date) + "\"";
        out = re.execute(command);
        System.out.println(out);

        command = "hadoop jar \"" + mapreduceJar + "\" " + mapreduceClass + " -i \"" + mapreduceInput + "/" + toDateString(date) + "\" -o \"" + mapreduceOutput + "/" + toDateString(date) + "\"";
        out = re.execute(command);
        System.out.println(out);

        System.out.println("执行完毕。");
    }

    public static void exportData() throws Exception {
        exportData(new Date());
    }

    public static void exportData(Date date) throws Exception {
        System.out.println("* 启动 Sqoop 导出程序，将 页面访问量统计结果 导出到 MySQL 中");
        pageViewService.dropTable();
        pageViewService.createTable();
        RemoteExecute re = RemoteExecute.getInstance();
        String out;
        String command = "sqoop export" + " \\" +
                "--connect \"" + datasourceUrl + "\"" + " \\" +
                "--username " + datasourceUsername + " \\" +
                "--password " + datasourcePassword + " \\" +
                "--table " + "page_view" + " \\" +
                "--input-fields-terminated-by " + "'\\t'" + " \\" +
                "--export-dir " + mapreduceOutput + "/" + toDateString(date) + " \\" +
                "-m " + "1";
        out = re.execute(command);
        System.out.println(out);
        System.out.println("执行完毕。");
    }
}
